package com.hgw.rocketmqtemplatedemo.controller;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * Description: 消息生产者demo
 *
 * @author LinHuiBa-YanAn
 * @date 2023/2/15 19:35
 */
@RestController
@RequestMapping("send")
public class SendMsgController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 普通消息无返回值，只负责发送消息⽽不等待服务器回应且没有回调函数触发
     * - 参数一：topicName:tags
     * - 参数二：消息体
     */
    @GetMapping("/base")
    public void sendBaseMsg() {
        rocketMQTemplate.convertAndSend("Demo:base", "普通消息测试");
    }

    /**
     * 过滤消息
     */
    @GetMapping("/filter")
    public void sendFilterMsg() {
        for (int i = 0; i < 10; i++) {
            HashMap<String, Object> harder = new HashMap<>(1);
            harder.put("a", String.valueOf(i));
            rocketMQTemplate.convertAndSend("Demo:filter", "过滤消息测试" + i, harder);
        }
    }

    /**
     * 同步消息有返回值SendResult，等到消息发送成功后才算结束。
     */
    @GetMapping("/sync")
    public String sendSyncMsg() {
        SendResult result = rocketMQTemplate.syncSend("Demo:sync", "同步消息测试");
        return JSONObject.toJSONString(result);
    }

    /**
     * 延时消息
     */
    @GetMapping("/delay")
    public void sendDelayMsg() {
        Message<String> msg = MessageBuilder.withPayload("延时消息测试").build();
        rocketMQTemplate.syncSend("Demo:delay", msg, 200, 3);
    }

    /**
     * 异步消息无返回值，需要传入回调类。无需等待消息是否发送成功。
     */
    @GetMapping("/async")
    public void sendAsyncMsg() {
        rocketMQTemplate.asyncSend("Demo:async", "异步消息测试", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("异步消息发送失败");
            }
        });
    }

    /**
     * 批量消息
     */
    @GetMapping("/batch")
    public void sendOneMsg() {
        ArrayList<Message<String>> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            messages.add(MessageBuilder.withPayload("批量消息" + (i + 1)).build());
        }
        rocketMQTemplate.syncSend("Demo:batch", messages);
    }

    /**
     * 顺序消息
     */
    @GetMapping("/order")
    public void sendOrderMsg() {
        // 自定义选择队列的规则，指定同一个队列
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, org.apache.rocketmq.common.message.Message msg, Object arg) {
                return mqs.get(0);
            }
        });
        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.syncSendOrderly("Demo:order", "同步发送顺序消息" + i, 0 + "");
        }

        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.asyncSendOrderly("Demo:order", "异步发送顺序消息" + i, 0 + "", new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    return;
                }

                @Override
                public void onException(Throwable e) {
                    return;
                }
            });
        }

        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.sendOneWayOrderly("Demo:order", "单向发送顺序消息" + i, 0 + "");
        }

    }


    /**
     * 发送事务消息测试
     */
    @GetMapping("/tx")
    public void sendTransactionMsg() {
        for (int i = 1; i <= 10; i++) {
            // 发送指定事务id的消息
            Message msg = MessageBuilder.withPayload("事务消息" + i).setHeader(RocketMQHeaders.KEYS, i).build();
            TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("Demo:tx", msg, null);
        }
    }
}
